package com.yl.util;

import cn.hutool.json.JSONUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.yl.constant.CdcOperation;
import com.yl.constant.CdcType;
import com.yl.constant.Const;
import com.yl.entity.cdc.MysqlCdcEntity;
import com.yl.flink.processor.MysqlCdcParseFunc;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

/**
 * @author wlf
 * @since 2022/8/10
 */
public class MysqlCdcUtil {

    /**
     * @param params 环境配置获取工具
     * @return mysql-cdc配置流
     */
    public static SingleOutputStreamOperator<MysqlCdcEntity> getConfigStream(StreamExecutionEnvironment env, ParameterTool params) {
        // 忽略非法反射警告
        JdkUtil.disableWarning();
        return env
                .fromSource(getSource(params), WatermarkStrategy.noWatermarks(), Const.FLINK_MYSQL_CDC_SOURCE)
                .setParallelism(getSourceParallelism(params))
                .process(new MysqlCdcParseFunc());

    }

    /**
     * @param params 环境配置获取工具
     * @return mysql-cdc数据源读取并行度
     */
    private static int getSourceParallelism(ParameterTool params) {
        String tables = params.get(Const.MYSQL_CDC_TABLES);
        return tables.split(Const.PTN_COMMA).length;
    }

    /**
     * @param params 环境配置获取工具
     * @return mysql-cdc数据源
     */
    private static MySqlSource<String> getSource(ParameterTool params) {
        String db = SUtil.getDB(params.get(Const.MYSQL_URL));
        String tables = SUtil.fmtTables(db, params.get(Const.MYSQL_CDC_TABLES));
        return MySqlSource.<String>builder()
                .port(3306)
                .hostname(SUtil.getHost(params.get(Const.MYSQL_URL)))
                .databaseList(db)
                .tableList(tables)
                .username(params.get(Const.MYSQL_USERNAME))
                .password(params.get(Const.MYSQL_PASSWORD))
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
    }

    /**
     * @param cdcJson  cdc输出的json串
     * @param srcTable 解析哪个表
     * @param tClass   解析成的对象
     * @param <T>      表对象
     * @return 表对象
     */
    public static <T> MysqlCdcEntity<T> parse(String cdcJson, String srcTable, Class<T> tClass) {
        // 获取操作类型
        CdcOperation cdcOperation = CdcOperation.getByOp(getFromJson(cdcJson, Const.S_OP));
        // 获取表名
        String table = getFromJson(getFromJson(cdcJson, Const.S_SOURCE), Const.S_TABLE);
        if (table.equals(srcTable)) {
            MysqlCdcEntity<T> cdcEntity = new MysqlCdcEntity<>();
            cdcEntity.setCdcOperation(cdcOperation);
            cdcEntity.setEntity(JSONUtil.toBean(getFromJson(cdcJson, Const.S_AFTER), tClass));
            return cdcEntity;
        } else {
            return null;
        }
    }

    /**
     * 从json字符串中获取指定key的值
     */
    private static String getFromJson(String json, String key) {
        return JSONUtil.parseObj(json).get(key).toString();
    }
}
